/*
* This file is a part of the open-eBackup project.
* This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
* If a copy of the MPL was not distributed with this file, You can obtain one at
* http://mozilla.org/MPL/2.0/.
*
* Copyright (c) [2024] Huawei Technologies Co.,Ltd.
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
*/
#include "BackupHandler.h"

#include <map>
#include <thread>
#include <iostream>
#include <fstream>
#include <functional>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#ifndef WIN32
#include <unistd.h>
#endif
#ifdef WIN32
#include <windows.h>
#endif
#include "BackintConst.h"
#include "common/File.h"
#include "common/JsonHelper.h"

using namespace backint;
using namespace Module;

namespace {
const int WAIT_TIMOUT_MS = 5;
const int MAX_BACKUP_THREDS = 50;
const std::string BACKUP_THREAD_KEY = "BackupHandler";

const std::string BACKUP_SUCCESS = "#SAVED";
const std::string BACKUP_ERROR = "#ERROR";
const std::string VERSION_1_0 = "1.0";
const int NO_DATA_SIZE = -1;
const int READ_WRITE_DATA_BUFFER = 4096 * 1024;
}  // namespace

class XTaskBackup : public ExecutableItem {
public:
    XTaskBackup(const std::string &src, const std::string &dst, const std::string &backupId, int64_t size)
        : m_strFile(src), m_dstFile(dst), m_backupId(backupId), m_dataSize(size)
    {
        m_result = 0;
    }

    virtual ~XTaskBackup()
    {}
    virtual void Exec();
    std::string GetSrcFile()
    {
        return m_strFile;
    }

private:
    std::string m_strFile;
    std::string m_dstFile;
    std::string m_backupId;
    int64_t m_dataSize = 0;
};

void XTaskBackup::Exec()
{
    std::ifstream inFile(m_strFile.c_str(), std::ios::in | std::ios::binary);
    if (!inFile.is_open()) {
        ERRLOG("Open backup file failed, inputFile:%s, errno:%d", m_strFile.c_str(), GetLastError());
        m_result = -1;
        std::cout << BACKUP_ERROR << " " << m_strFile << std::endl;
        return;
    }

    std::ofstream outFile(m_dstFile.c_str(), std::ios::out | std::ios::binary);
    if (!outFile.is_open()) {
        ERRLOG("Open dest file failed, outputFile:%s, errno:%d", m_dstFile.c_str(), GetLastError());
        inFile.close();
        m_result = -1;
        std::cout << BACKUP_ERROR << " " << m_strFile << std::endl;
        return;
    }

    std::streamsize totalSize = 0;
    char readBuffer[READ_WRITE_DATA_BUFFER + 1] = {0};
    while (!inFile.eof()) {
        inFile.read(readBuffer, READ_WRITE_DATA_BUFFER);
        std::streamsize bytesRead = inFile.gcount();
        totalSize += bytesRead;
        outFile.write(readBuffer, bytesRead);
    }
    inFile.close();
    outFile.close();
    m_result = 0;
    printf("%s %s %s %zu\n", BACKUP_SUCCESS.c_str(), m_backupId.c_str(), m_strFile.c_str(), totalSize);
    DBGLOG("Exit m_backupID: %s, inputFile:%s, size: %lu", m_backupId, m_strFile.c_str(), totalSize);
}

BackupHandler::BackupHandler()
{
    m_backupFunMap.insert(std::make_pair<std::string, BackupFunc>(
        "COMPLETE", std::bind(&BackupHandler::DataBackup, this, std::placeholders::_1, std::placeholders::_2)));
    m_backupFunMap.insert(std::make_pair<std::string, BackupFunc>(
        "DIFFERENTIAL", std::bind(&BackupHandler::DataBackup, this, std::placeholders::_1, std::placeholders::_2)));
    m_backupFunMap.insert(std::make_pair<std::string, BackupFunc>(
        "INCREMENTAL", std::bind(&BackupHandler::DataBackup, this, std::placeholders::_1, std::placeholders::_2)));
    m_backupFunMap.insert(std::make_pair<std::string, BackupFunc>(
        "METADATA", std::bind(&BackupHandler::DataBackup, this, std::placeholders::_1, std::placeholders::_2)));
    m_backupFunMap.insert(std::make_pair<std::string, BackupFunc>(
        "INITIALVOLUME", std::bind(&BackupHandler::DataBackup, this, std::placeholders::_1, std::placeholders::_2)));
}

BackupHandler::~BackupHandler()
{
    ThreadPoolFactory::DestoryThreadPool(BACKUP_THREAD_KEY);
}

uint32_t BackupHandler::ExecBackup(const StuParam &inParam, const PlugParams &plugParam)
{
    LOGGUARD("");
    GetBackupParam(inParam, plugParam);
    StuFileInfo stFileInfo;
    if (ReadPipeAndFile(inParam.m_inputFile, stFileInfo) != SUCCESS) {
        ERRLOG("Failed to read file, jobId=%s, subJobId=%s.", m_jobId.c_str(), m_subJobId.c_str());
        return FAILED;
    }
    auto iter = m_backupFunMap.find(inParam.m_backupType);
    if (iter != m_backupFunMap.end()) {
        return iter->second(inParam, stFileInfo);
    }
    ERRLOG("Undefined backup type %s.", inParam.m_backupType.c_str());
    return FAILED;
}

uint32_t BackupHandler::DataBackup(const StuParam &inParam, const StuFileInfo &stFileInfo)
{
    LOGGUARD("");
    return ReadWriteDataForOra(inParam, stFileInfo);
}

uint32_t BackupHandler::ReadPipeAndFile(const std::string &inputFile, StuFileInfo &stFileInfo)
{
    if (GetBackupList(inputFile, stFileInfo) != SUCCESS) {
        ERRLOG("Failed to read pipe info.");
        return FAILED;
    }

    if (stFileInfo.fileData.empty()) {
        ERRLOG("Pipe info and file info is empty.");
        return FAILED;
    }
    INFOLOG("The files size:", stFileInfo.fileData.size());
    return SUCCESS;
}

void BackupHandler::GetBackupParam(const StuParam &inParam, const PlugParams &plugParam)
{
    m_jobId = plugParam.m_jobId;
    m_subJobId = plugParam.m_subJobId;
    m_cachePath = plugParam.m_cachePath;
    m_dataPath = plugParam.m_dataPath;
    m_outputFile = inParam.m_outFile;
    m_backupID = inParam.m_dtBackupId;
    uint32_t cpuNum = GetCpuNum();
    if (plugParam.m_channels > 0 && (uint32_t)plugParam.m_channels <= cpuNum) {
        m_channels = plugParam.m_channels;
    }
}

void BackupHandler::CheckTaskResult(size_t &nJobs, bool &result)
{
    std::shared_ptr<ExecutableItem> treadPoolRes;
    while (m_jsPtr->Get(treadPoolRes, true, WAIT_TIMOUT_MS)) {
        --nJobs;
        XTaskBackup *xTask = (XTaskBackup *)treadPoolRes.get();
        if (xTask == nullptr) {
            result = false;
            ERRLOG("xTask is null");
            break;
        }

        if (xTask->Result() != 0) {
            result = false;
            ERRLOG("Task is failed, backup file:%s", xTask->GetSrcFile().c_str());
            break;
        }
        ++m_backedCount;
        DBGLOG("Task is success[%d/%d], backup file:%s", m_backedCount, m_backupTotal, xTask->GetSrcFile().c_str());
    }
}

void BackupHandler::BackupTaskInner(size_t &nJobs, bool &result)
{
    std::shared_ptr<ExecutableItem> treadPoolRes;
    while (nJobs) {
        if (!m_jsPtr->Get(treadPoolRes, true)) {
            ERRLOG("Get task failed.");
            result = false;
            break;
        }

        --nJobs;
        XTaskBackup *xTask = (XTaskBackup *)treadPoolRes.get();
        if (xTask == nullptr) {
            ERRLOG("xTask is null.");
            result = false;
            break;
        }

        if (xTask->Result() != 0) {
            result = false;
            ERRLOG("Task is failed, backup file:%s", xTask->GetSrcFile().c_str());
        }
        ++m_backedCount;
        DBGLOG("Task is success[%d/%d], backup file:%s", m_backedCount, m_backupTotal, xTask->GetSrcFile().c_str());
    }
}

bool BackupHandler::ExecutTask(size_t nJobs, const StuFileData &fileData)
{
    std::string fileName = GetFileName(fileData.m_filePath);
    std::string outputFile = m_dataPath + PATH_SEP_CHAR + fileName;
    std::shared_ptr<ExecutableItem> task(
        new (std::nothrow) XTaskBackup(fileData.m_filePath, outputFile, m_backupID, fileData.m_dataLength));
    if (task == nullptr) {
        ERRLOG("task is nullptr");
        return false;
    }

    DBGLOG("Put task backup file:%s", fileData.m_filePath.c_str());
    if (!m_jsPtr->Put(task)) {
        ERRLOG("Put failed, backup file:%s", fileData.m_filePath.c_str());
        return false;
    }
    return true;
}

uint32_t BackupHandler::ReadWriteDataForOra(const StuParam &inParam, const StuFileInfo &fileInfo)
{
    LOGGUARD("");
    m_jsPtr = std::make_shared<JobScheduler>(*ThreadPoolFactory::GetThreadPoolInstance(BACKUP_THREAD_KEY, m_channels));
    if (m_jsPtr == nullptr) {
        ERRLOG("m_jsPtr malloc failed");
        return FAILED;
    }

    size_t nJobs = 0;
    bool result = true;
    size_t index = 0;
    m_backupTotal = (uint32_t)(fileInfo.fileData.size());
    INFOLOG("The backup channels: %d, backup file size: %d", m_channels, m_backupTotal);
    for (; index < m_backupTotal;) {
        if (nJobs < m_channels) {
            if (!ExecutTask(nJobs, fileInfo.fileData[index])) {
                result = false;
                break;
            }
            ++index;
            ++nJobs;
        }
        CheckTaskResult(nJobs, result);
        if (!result) {
            break;
        }
    }
    DBGLOG("nJobs: %d", nJobs);
    BackupTaskInner(nJobs, result);
    INFOLOG("Exit ReadWriteDataForOra nJobs: %d", nJobs);
    return result ? SUCCESS : FAILED;
}